package com.atguigu.gmall;

import com.alibaba.fastjson.JSONObject;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.atguigu.gmall.contants.GmallConstants;
import com.google.protobuf.ByteString;
import com.google.protobuf.InvalidProtocolBufferException;

import java.net.InetSocketAddress;
import java.util.List;
import java.util.Random;

/**
 * Date:2021/3/20
 * Author:csw
 * Description:
 */
public class MyCanalClient {
    public static void main(String[] args) throws InterruptedException, InvalidProtocolBufferException {
        CanalConnector canalConnector = CanalConnectors.newSingleConnector(new InetSocketAddress("hadoop102", 11111),
                "example", "", "");
        canalConnector.connect();
        canalConnector.subscribe("gmall1026.*");
        while (true){
            Message message = canalConnector.get(100);
            if(message.getId()==-1){
                Thread.sleep(3000);
                System.out.println("无数据");
                continue;
            }
            System.out.println(message);
            List<CanalEntry.Entry> entries = message.getEntries();

            for (CanalEntry.Entry entry : entries) {
                if (entry.getEntryType().equals(CanalEntry.EntryType.ROWDATA)) {
                    ByteString storeValue = entry.getStoreValue();
                    String tableName = entry.getHeader().getTableName();
                   // System.out.println(storeValue);
                    handleData(storeValue,tableName);
                }
            }


        }
    }

    private static void handleData(ByteString storeValue, String tableName) throws InvalidProtocolBufferException {
        CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(storeValue);
        if("order_info".equals(tableName) && rowChange.getEventType().equals(CanalEntry.EventType.INSERT)){
               cancalToKafka(rowChange,GmallConstants.KAFKA_TOPIC_ORDER_INFO);

        }else if("order_detail".equals(tableName) && rowChange.getEventType().equals(CanalEntry.EventType.INSERT)){
            cancalToKafka(rowChange,GmallConstants.KAFKA_TOPIC_ORDER_DETAIL);
        }else if("user_info".equals(tableName) && (rowChange.getEventType().equals(CanalEntry.EventType.INSERT) || rowChange.getEventType().equals(CanalEntry.EventType.UPDATE)) ){
            cancalToKafka(rowChange,GmallConstants.KAFKA_TOPIC_USER_INFO);
        }
    }

    public static void cancalToKafka(CanalEntry.RowChange rowChange,String topic){

        List<CanalEntry.RowData> rowDatasList = rowChange.getRowDatasList();
        for (CanalEntry.RowData rowData : rowDatasList) {
            JSONObject jsonObject = new JSONObject();
            List<CanalEntry.Column> columnsList = rowData.getAfterColumnsList();
            for (CanalEntry.Column column : columnsList) {
                jsonObject.put(column.getName(),column.getValue());
            }
            int sleeptime = new Random().nextInt(5) * 1000;
            try {
                Thread.sleep(sleeptime);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println(jsonObject);
            MyProducer.send(topic,jsonObject.toString());
        }
    }


}
